package com.to8to.autotask.util;

import java.net.InetAddress;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.List;

import javax.sql.rowset.CachedRowSet;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.sun.rowset.CachedRowSetImpl;
import com.to8to.autotask.rpc.SendMsg2ES;
import com.to8to.crm.common.CrmDataSource;
import com.to8to.es.mq.EsWriter;
import com.to8to.es.util.LogUtil;

/**
 * DBUtil.java
 * 作 者：atom.liu
 * 版 本：v2.0
 * 日 期：2014年12月3日
 * 描 述：DB操作工具类
 */
public class DBUtil
{
    /**
     * 数据源
     */
    private static CrmDataSource ds = null;

    private static Logger logger = LoggerFactory.getLogger(DBUtil.class);

    private static String IP;

    static
    {
        @SuppressWarnings("resource")
        ApplicationContext context = new ClassPathXmlApplicationContext("spring-config.xml");
        ds = (CrmDataSource) context.getBean("dataSource");

        InetAddress host = Utils.getHostAddress();
        IP = host == null ? "" : host.getHostAddress();
        logger.info("Get Host IP : " + IP);
    }

    /**
     * 获取数据库连接，默认设置自动提交
     */
    public static Connection getConnection() throws SQLException
    {
        return getConnection(true);
    }

    /**
     * 获取数据库连接
     * @param isAutoCommit 是否自动提交。
     */
    public static Connection getConnection(boolean isAutoCommit) throws SQLException
    {
        Connection connect = ds.getConnection();
        connect.setAutoCommit(isAutoCommit);
        return connect;
    }

    /**
     * 获取数据库连接，使用try-catch处理，异常则返回null
     */
    public static Connection getConnectionWithTryCatch()
    {
        try
        {
            return DBUtil.getConnection();
        }
        catch (SQLException e)
        {
            logger.error("获取连接失败！ {}\n{}", e, LogUtil.logStackTrace(e));
            return null;
        }
    }

    /**
     * 回滚
     */
    public static void rollback(Connection connect)
    {
        try
        {
            if (null != connect && !connect.isClosed())
            {
                connect.rollback();
                logger.info("事务回滚 ...");
            }
        }
        catch (SQLException e)
        {
            logger.error("回滚出现异常！ {}\n{}", e, LogUtil.logStackTrace(e));
        }
    }

    /**
     * 将分配的SQL语句刷新到数据库中
     * @param connect 数据库连接
     * @param sql 要执行的sql语句
     * @param sendMsg 通知给ES的消息对象
     */
    public static void refresh2DB(Connection connect, String sql, int taskType, SendMsg2ES sendMsg) throws SQLException
    {
        List<String> list = Arrays.asList(sql);
        refresh2DB(connect, list, taskType, sendMsg);
    }

    /**
     * 将分配的SQL语句刷新到数据库中
     * @param connect 数据库连接
     * @sqls 要执行的sql语句集合
     * @senMsg 通知给ES的消息对象
     */
    public static void refresh2DB(Connection connect, List<String> sqls, int taskType, SendMsg2ES sendMsg) throws SQLException
    {
        if (Utils.isNullOrEmpty(sqls))
        {
            logger.info("DBUtil.refresh2DB(): 本次没有要进行更新的Sql语句！");
            return;
        }

        logger.info("开始将信息同步到数据库...sqls.size : " + sqls.size());

        if (sqls.size() > 5000)
        {
            refreshBigData(connect, sqls, taskType, sendMsg);
            return;
        }

        Statement stat = connect.createStatement();
        StringBuilder sbLog = new StringBuilder();
        for (int i = 0; i < sqls.size(); i++)
        {
            String sql = sqls.get(i);
            stat.addBatch(sql);
            sbLog.append("SQL: ").append(sql).append("\r\n");
            if (i % 200 == 0 && i > 0)
            {
                logger.info(sbLog.toString());
                sbLog = new StringBuilder();
            }
            else if (i == sqls.size() - 1)
            {
                logger.info(sbLog.toString());
            }
        }
        int[] n = stat.executeBatch();
        if (!connect.getAutoCommit())
            connect.commit();
        logger.info(String.format("同步完成， 共执行 %s 条sql语句。每条更新记录数：%s", n.length, Arrays.toString(n)));
        if (n.length == 0)
            return;
        DBUtil.releaseSource(null, stat, null);
        DBUtil.recordAutoTaskLog(connect, taskType, sqls, false);
        if (sendMsg != null)
            sendMsg.sendMsg();
    }

    /**
     * 当数据量比较大时，批量提交，批量打印日志
     */
    private static void refreshBigData(Connection connect, List<String> sqls, int taskType, SendMsg2ES sendMsg) throws SQLException
    {
        Statement stat = connect.createStatement();
        StringBuilder sbLog = new StringBuilder();
        for (int i = 0; i < sqls.size(); i++)
        {
            String sql = sqls.get(i);
            stat.addBatch(sql);
            if (i % 5000 == 0 && i > 0) //5000条批量提交一次
            {
                int[] n = stat.executeBatch();
                logger.info(String.format("同步完成，本次执行 %s 条sql语句。", n.length));
            }
            else if (i == sqls.size() - 1)
            {
                int[] n = stat.executeBatch();
                logger.info(String.format("同步完成，本次执行 %s 条sql语句。", n.length));
            }
            sbLog.append("SQL: ").append(sql).append("\r\n");
            if (i % 200 == 0 && i > 0) //200条打印一次日志
            {
                logger.info(sbLog.toString());
                sbLog = new StringBuilder();
            }
            else if (i == sqls.size() - 1)
            {
                logger.info(sbLog.toString());
            }
        }
        if (!connect.getAutoCommit())
            connect.commit();
        DBUtil.releaseSource(null, stat, null);
        int count = (sqls.size() + 4999) / 5000;
        for (int i = 0; i < count; i++)
        {
            int end = (i == count - 1) ? sqls.size() : (i + 1) * 5000;
            List<String> subList = sqls.subList(i * 5000, end);
            DBUtil.recordAutoTaskLog(connect, taskType, subList, false);
        }

        if (sendMsg != null)
            sendMsg.sendMsg();
    }

    /**
     * 今天是已执行过某项自动任务的次数
     * @param connect
     * @param taskType
     * @return 
     */
    public static int execTaskCountsAtToday(Connection connect, int taskType)
    {
        int result = -1;
        String sql = "select count(*) from to8to_autotask_log where taskType = ? and ctime >= ? and ctime < ?";
        try
        {
            PreparedStatement stat = connect.prepareStatement(sql);
            stat.setInt(1, taskType);
            stat.setTimestamp(2, new Timestamp(Utils.getDayTimePoint(0) * 1000L));
            stat.setTimestamp(3, new Timestamp(Utils.getDayTimePoint(1) * 1000L));
            ResultSet rs = stat.executeQuery();
            while (rs.next())
                result = rs.getInt(1);
            DBUtil.releaseSource(rs, stat, null);
        }
        catch (SQLException e)
        {
            logger.error("DBUtil.recordAutoTaskLog()异常！ {}\n{}", e, LogUtil.logStackTrace(e));
        }
        if (result > 0)
            logger.error(String.format("今天已经执行过[TaskType %s ]的任务！本次将不再执行！", taskType));
        return result;
    }

    public static void recordAutoTaskLog(Connection connect, int taskType, List<String> sqls, boolean isPrintSqls)
    {
        String sql = "insert into to8to_autotask_log(tasktype,sqltext,ctime,ip) values(?,?,?,?)";
        try
        {
            PreparedStatement stat = connect.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
            String sqlText = Utils.collect2Str(sqls, "#\r\n");
            if (isPrintSqls)
                logger.info(sqlText);
            stat.setInt(1, taskType);
            stat.setString(2, sqlText);
            stat.setTimestamp(3, new Timestamp(System.currentTimeMillis()));
            stat.setString(4, IP);
            stat.executeUpdate();
            ResultSet rs = stat.getGeneratedKeys();
            int id = -1;
            while (rs.next())
                id = rs.getInt(1);
            if (!connect.getAutoCommit())
                connect.commit();
            logger.info("DBUtil.recordAutoTaskLog Succ! id : " + id);
            DBUtil.releaseSource(rs, stat, null);
        }
        catch (SQLException e)
        {
            logger.error("DBUtil.recordAutoTaskLog()异常！ {}\n{}", e, LogUtil.logStackTrace(e));
        }
    }

    /**
     * 发送消息给es，刷新es
     */
    public static void sendUpdateMsg2ES(String tableName, List<String> ids)
    {
        EsWriter.sendUpdateMsg(tableName, ids);
    }

    /**
     * 获取某表某行数据某字段值
     * @param tableName DB表名
     * @param fieldName 字段名
     * @param id 数据行id
     * @return 字段值
     */
    public static Object getField(String tableName, String fieldName, String idField, String id)
    {
        Object val = null;
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;
        try
        {
            conn = getConnection();
            String sql = "select " + fieldName + " from " + tableName + " where " + idField + " = " + id;
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);
            if (rs.next())
                val = rs.getObject(1);
        }
        catch (SQLException e)
        {
            logger.error("get field 失败 !exception : {}\n{}", e, LogUtil.logStackTrace(e));
        }
        finally
        {
            releaseSource(rs, stmt, conn);
        }
        return val;
    }

    /**
     * 更新字段值
     * @param tableName 表名
     * @param fieldName 要更新的字段名
     * @param newValue 新值
     * @param idField 主键id
     * @param id id值
     * @return
     */
    public static int updateField(String tableName, String fieldName, String newValue, String idField, String id)
    {
        Connection conn = null;
        Statement stmt = null;
        try
        {
            conn = getConnection();
            StringBuilder sql = new StringBuilder();
            sql.append("update ").append(tableName).append(" set ").append(fieldName).append(" = ").append(newValue);
            sql.append(" where ").append(idField).append(" = ").append(id);
            stmt = conn.createStatement();
            return stmt.executeUpdate(sql.toString());
        }
        catch (SQLException e)
        {
            e.printStackTrace();
            logger.error("get field 失败 !exception : {}\n{}", e, LogUtil.logStackTrace(e));
        }
        finally
        {
            releaseSource(null, stmt, conn);
        }
        return 0;
    }

    /**
     * 执行查询sql语句。
     */
    @Deprecated
    public static CachedRowSet executeQuery(Connection connect, String sql)
    {
        logger.info("DBUtil.executeQuery() Sql: " + sql);
        Statement stat = null;
        ResultSet rs = null;
        try
        {
            CachedRowSet rowSet = new CachedRowSetImpl();
            stat = connect.createStatement();
            rs = stat.executeQuery(sql);
            rowSet.populate(rs);
            return rowSet;
        }
        catch (SQLException e)
        {
            logger.error("DBUtil.executeQuery()异常！ {}\n{}", e, LogUtil.logStackTrace(e));
            return null;
        }
        finally
        {
            releaseSource(rs, stat, null);
        }
    }

    /**
     * 释放数据库连接资源 
     */
    public static void releaseSource(ResultSet result, Statement stat, Connection connect)
    {
        if (result != null)
        {
            try
            {
                if (!result.isClosed())
                    result.close();
            }
            catch (SQLException e)
            {
                logger.error("关闭ResultSet异常！{} \n {}", e, LogUtil.logStackTrace(e));
            }
        }
        if (stat != null)
        {
            try
            {
                if (!stat.isClosed())
                    stat.close();
            }
            catch (SQLException e)
            {
                logger.error("关闭Statement异常！{} \n {}", e, LogUtil.logStackTrace(e));
            }
        }

        if (connect != null)
        {
            try
            {
                if (!connect.isClosed())
                    connect.close();
            }
            catch (SQLException e)
            {
                logger.error("关闭Connection异常！{} \n {}", e, LogUtil.logStackTrace(e));
            }
        }
    }
}
